package com.qyer.search.storm.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
import backtype.storm.Config;
import com.qyer.search.storm.logging.ErrorLogger;
import com.qyer.search.storm.logging.InfoLogger;
import com.qyer.search.storm.logging.TimeCostLogger;
import com.qyer.search.storm.logging.WarnLogger;
import com.qyer.search.storm.util.CommonCityService;
import com.qyer.search.storm.util.CommonCountryService;
import com.qyer.search.storm.util.ConfigUtils;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import backtype.storm.utils.TupleUtils;

import java.sql.PreparedStatement;
import java.util.*;
import java.sql.Connection;

/**
 * Created by tangzheng on 16/1/8.
 */
public class JDBCBolt implements IRichBolt {

  private static final long serialVersionUID = -1L;
  protected int _inCount;
  protected int _inCount2;
  protected String _boltPrefix;
  protected DateTime _boltDateTime;
  protected String _boltTimeStr;
  protected long _boleMilliSecond;
  protected Connection _conn;
  protected PreparedStatement _pstmt;
  protected PreparedStatement _pstmt2;
  protected Map<String, Object> _config;
  protected OutputCollector _outputCollector;
  protected TopologyContext _topologyContext;
  protected int _taskId;
  protected String _componentId;
  protected String _prefix;
  protected Map _sqlConfig;

  protected InfoLogger dataLog;
  protected InfoLogger infoLogger;
  protected InfoLogger dbInfoLogger;
  protected InfoLogger tickInfoLogger;
  protected InfoLogger processLogger;
  protected ErrorLogger errorLogger;
  protected WarnLogger failLogger;
  protected WarnLogger successLogger;
  protected TimeCostLogger timeCostLogger;



  public JDBCBolt(String boltPrefix, int inCount, int inCount2, Map sqlConfig){
    _boltPrefix = boltPrefix;
    _inCount = inCount;
    _inCount2 = inCount2;
    _boleMilliSecond = 0;
    _sqlConfig = sqlConfig;
    _taskId = -1;
    _componentId = "undefined".toUpperCase();
    _prefix = "TaskId:-1,ComponentId:UNDEFINED";
  }

  @Override public void prepare(Map map, TopologyContext topologyContext,
                                OutputCollector outputCollector){

    String directory = "/Data/search/JDBCBolt/";
    String filename = directory + topologyContext.getStormId()+"/"
      + topologyContext.getThisComponentId() +"/"+topologyContext.getThisTaskId();
    String loggername = JDBCBolt.class.getSimpleName() + "@" + topologyContext.getThisTaskId();
    InfoLogger tmpLog = new InfoLogger(filename + "_tmp", loggername + "_tmp");
    dataLog = new InfoLogger(filename + "_data", loggername + "_data");
    infoLogger = new InfoLogger(filename + "_info", loggername + "_info");
    errorLogger = new ErrorLogger(filename +"_error", loggername + "_error");
    timeCostLogger = new TimeCostLogger(filename + "_timecost", loggername + "_timecost");
    successLogger = new WarnLogger(filename + "_success", loggername + "_success");
    failLogger = new WarnLogger(filename + "_fail", loggername + "_fail");
    dbInfoLogger = new InfoLogger(filename + "_db", loggername + "_db");
    tickInfoLogger = new InfoLogger(filename + "_tick", loggername + "_tick");
    processLogger = new InfoLogger(filename + "_process", filename + "_process");


    _topologyContext = topologyContext;
    _outputCollector = outputCollector;
    _config = (Map<String, Object>)map;
    _taskId = topologyContext.getThisTaskId();
    _componentId = topologyContext.getThisComponentId();
    _prefix = _boltPrefix + ",TaskId:"+ _taskId + ",ComponentId:" + _componentId;
    //currentTime(_prefix + "Bolt prepare");

    try {
      _conn = ConfigUtils.getConnection((Map<String, Object>) _sqlConfig, _boltPrefix);
      //LOG.info("{} DB init successfully!", _boltPrefix);
      dbInfoLogger.info("{} DB init successfully!", _boltPrefix);
    }catch (Exception e){
      //LOG.error("{} Connect DB failed.", _boltPrefix);
      dbInfoLogger.info("{} Connect DB failed.", _boltPrefix);
    }

    try {
      _pstmt = ConfigUtils.getPrepareStmt((Map<String, Object>) map, _boltPrefix, this._conn);
      _pstmt2 = ConfigUtils.getPrepareStmt2((Map<String, Object>) map, _boltPrefix, this._conn);
    }catch (Exception e){
      //LOG.error("{} get PreparedStatement failed.", _boltPrefix);
      dbInfoLogger.info("{} get PreparedStatement failed.", _boltPrefix);
    }


  }

  @Override public void execute(Tuple tuple) {

  }

  @Override public void cleanup() {
    //currentTime(_prefix + "cleanup,");
    try {
      if( _pstmt != null)
        _pstmt.close();
      if(_pstmt2 != null)
        _pstmt2.close();
      if(_conn != null)
        _conn.close();
      dbInfoLogger.info("mysqlconnection closed");
      //LOG.info("mysqlconnection closed");
    }catch (Exception e){
      dbInfoLogger.info(_prefix + e.getMessage());
    }
  }

  @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

  }

  @Override public Map<String, Object> getComponentConfiguration() {

    //currentTime(_prefix + ",getComponentConfiguration");

    Config conf = new Config();
    conf.put(conf.TOPOLOGY_TICK_TUPLE_FREQ_SECS,10);
    return conf;

  }

  public void currentTime(String tag){
    long _start = _boleMilliSecond;
    _boltDateTime = DateTime.now();
    _boleMilliSecond = _boltDateTime.getMillis();
    _boltTimeStr = _boltDateTime.toString(ConfigUtils.DATE_FORMAT);
  }

  public void checkDBConnection(){
    try {
      if((_conn == null) || (!_conn.isValid(2))){
        dbInfoLogger.info("mysqlconnection is invalid,rebuild follow.");
        Connection _tempConn = _conn;
        this._conn = ConfigUtils.getConnection(_sqlConfig, _boltPrefix);
        this._pstmt = ConfigUtils.getPrepareStmt(_config, _boltPrefix, _conn);
        this._pstmt2 = ConfigUtils.getPrepareStmt2(_config, _boltPrefix, _conn);
        if(_tempConn != null) {
          _tempConn.close();
          dbInfoLogger.info("close the past mysqlconnection");
        }
      }else {
        infoLogger.info("mysqlconnection is ok");
      }
    }catch (Exception e){
      e.printStackTrace();
      dbInfoLogger.info("mysqlconnection rebuild failed by {}.", e);
      dbInfoLogger.info("Exception:{},DB connection invalid.", e.getMessage());
    }
  }
}

